package com.whately.core.influxdb;

import com.whately.core.base.constant.DebugSwitcher;
import com.whately.core.base.helper.PubThreadFactory;
import lombok.Getter;
import org.apache.commons.lang3.StringUtils;
import org.influxdb.InfluxDB;
import org.influxdb.dto.Point;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 写入事件到InfluxData中
 */
@Component
public class InfluxDataUtil {

    private Logger log = LoggerFactory.getLogger(getClass());


    @Getter
    @Autowired
    private DebugSwitcher debugSwitcher;

    //线程池
    private ExecutorService executorService ;

    private final String rpName = "default";


    /**************************************************************************
     *   influx init
     *************************************************************************/


    @Value("${influxdb.enable:false}")
    private boolean enable;

    @Value("${influxdb.ip:0.0.0.0}")
    private String ip;

    @Value("${influxdb.user:user}")
    private String user;

    @Value("${influxdb.pwd:password}")
    private String pwd;



    private InfluxDBModel influxDBModel;

    @PostConstruct
    public void init() {

        if(!enable){
            log.info("influx db not start!");
            return ;
        }

        if(executorService==null){
            executorService = new ThreadPoolExecutor(1, 5, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000), new PubThreadFactory("influx-service-processor"));
        }

        influxDBModel = new InfluxDBModel();
        influxDBModel.setIp(ip);
        influxDBModel.setUser(user);
        influxDBModel.setPwd(pwd);

        //monitor, default
        InfluxDB influxDB = org.influxdb.InfluxDBFactory.connect(influxDBModel.getIp(), influxDBModel.getUser(), influxDBModel.getPwd());

        /***
         *  配置batch
         *   Flush every 100 Points, at least every 20s
         */
        influxDB.enableBatch(50, 5, TimeUnit.SECONDS);
        //influxDB.enableBatch(10, 100, TimeUnit.MILLISECONDS);

        influxDBModel.setInfluxDB(influxDB);

    }

    /**
     * 通过服务器名获取数据库
     */
    private InfluxDBModel getInfluxDB() {
        return influxDBModel;
    }



    /**************************************************************************
     *    influx write or query
     *************************************************************************/



    public void report(String database,Point points ) {
        this.report(database,points,rpName);
    }

    public void report( String database,Point points,String retentionPolicy ) {
        //没有model对象则退出
        if(null==this.getInfluxDB()){
            return ;
        }

        executorService.execute(()->{
            /**
             *  写入数据， 支持HTTP、UDP
             */
            boolean httpWriteFlag=true;
            InfluxDBModel influxDBModel=this.getInfluxDB();
            InfluxDB influxDB = influxDBModel.getInfluxDB();
            try {
                //HTTP还是UDP
                int udpPort= 0;
            /*udpPort = InfluxDBUdpEnum.getUdpPortUnderConfig(database,retentionPolicy);
            if(udpPort>0){
                //开启udp
                httpWriteFlag=false;
            }*/

                // httpWriteFlag=true;
                if (httpWriteFlag) {
                    influxDB.write(database, StringUtils.isBlank(retentionPolicy)?rpName:retentionPolicy,points);
                    //log.debug(" InfluxDataReporter write {}   with http success.", batchPoints);
                } else {
                    influxDB.write(udpPort, points);
                    //log.debug("InfluxDataReporter write {}  with udp success.", measurement, tags, database);
                }
            } catch (Exception e) {
                log.error("InfluxDataReporter write to influxdb error", e);
            }
        });

    }


    /**
     *  写入数据， 通过HTTP
     */
    /*public void reportByHttp(String dataSource, BatchPoints batchPoints  ) {
        InfluxDBModel influxDBModel=influxDBSingle.getInfluxDB();
        InfluxDB influxDB = influxDBModel.getInfluxDB();
        try {
            influxDB.write(batchPoints);
            //log.debug(" InfluxDataReporter write {}   with http success.", batchPoints);
        } catch (Exception e) {
            log.error("InfluxDataReporter write to influxdb error", e);
        }
    }*/

    public QueryResult queryResult(String dataSource, String command, String database){
        Query query = new Query(command, database);
        try {
            QueryResult queryResult = this.getInfluxDB()
                    .getInfluxDB().query(query);

            return queryResult;
        } catch (Exception e) {
            log.error("influxdb query failed! influxDBName:{},commond:{},database:{},exception{}",dataSource,command,database,e);
            return null;
        }
    }

    public Map<String,Object> queryWithCheckException(String influxDBName, String command, String database){
        Map<String,Object> map=new HashMap<String,Object>();
        String exceptionFlag="";
        Query query = new Query(command, database);
        try {
            QueryResult queryResult = this.getInfluxDB()
                    .getInfluxDB().query(query);
            map.put("queryResult",queryResult);
        } catch (Exception e) {
            exceptionFlag="1";
            log.error("influxdb query failed! influxDBName:{},commond:{},database:{},exception{}",influxDBName,command,database,e);
        }
        map.put("exceptionFlag",exceptionFlag);
        return map;
    }


}
